package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/24 11:39
 */
public class Dwd_13_DwdToolCouponPayApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_13_DwdToolCouponPayApp().init(2012, 2, "Dwd_13_DwdToolCouponPayApp");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        readOdsDb(tEnv, "Dwd_11_DwdToolCouponGetApp");
        
        
        // 过滤优惠券领用记录
        Table resultTable = tEnv.sqlQuery("select " +
                                              "data['id'], " +
                                              "data['coupon_id'], " +
                                              "data['user_id'], " +
                                              "date_format(data['get_time'],'yyyy-MM-dd') date_id, " +
                                              "data['used_time'] payment_time, " +
                                              "ts " +
                                              "from ods_db " +
                                              "where `table`='coupon_use' " +
                                              "and `type`='update' " +
                                              "and `old`['coupon_status']='1402' " +
                                              "and `data`['used_time'] is not null" );
        
        tEnv.executeSql("create table dwd_tool_coupon_payment ( " +
                            "id string, " +
                            "coupon_id string, " +
                            "user_id string, " +
                            "date_id string, " +
                            "payment_time string, " +
                            "ts string " +
                            ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_TOOL_COUPON_PAYMENT));
        resultTable.executeInsert("dwd_tool_coupon_payment");
        
    }
}
